#include "Redis.h"

enum class HandshakeType : int{
    INITIAL=0, // 初始状态，没有发送ping
    PING,   // 发送ping之后，期待pong
    REPLCONF_ONE, // 发送replconf capa之后，期待OK
    REPLCONF_TWO, // 发送replconf capa之后，期待OK
    PSYNC, // 发送psync之后，期待OK
    FINAL, // 完成握手
    ERROR, // 错误状态
};


void Redis::init_socket(){
    listen_fd_ = -1;
    listen_fd_ = socket(AF_INET, SOCK_STREAM, 0);

    if (listen_fd_ == -1) {
        perror("socket creation failed");
        exit(EXIT_FAILURE);
    }
    int opt = 1;
    if(setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))<0){
        cerr<<"setsockopt failed"<<endl;
        return;
    };
    struct sockaddr_in serv_addr;
    serv_addr.sin_family = AF_INET;

    serv_addr.sin_addr.s_addr = INADDR_ANY; // 只监听本地
    serv_addr.sin_port = htons(port);
    //cout<<"bind port "<<port<<endl;
    if (bind(listen_fd_, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }

    if (listen(listen_fd_, SOMAXCONN) == -1) {
        perror("listen failed");
        exit(EXIT_FAILURE);
    }

    epoll_.AddFd(listen_fd_, EPOLLIN);
    fcntl(listen_fd_, F_SETFL, fcntl(listen_fd_, F_GETFD, 0) | O_NONBLOCK);
};

void RedisSlave::init(){
    init_socket();
    this->redis_type=RedisType::REDIS_SLAVE;
    cout<<"Connecting to master "<<master_ip<<":"<<master_port<<endl;
    HandshakeType handshake_type=HandshakeType::INITIAL;
    char buffer[1024];int bytesRecv=0;
    string message;
    while(handshake_type!=HandshakeType::FINAL || handshake_type==HandshakeType::ERROR){
        switch (handshake_type)
        {
        case HandshakeType::INITIAL:
            {
                string message=array_string(vector<string>{"PING"});
                if (send(listen_fd_, message.c_str(),message.size(), 0) == -1) {
                    std::cerr << "Error: Failed to send message" << std::endl;
                    handshake_type=HandshakeType::ERROR;
                }else{
                    handshake_type=HandshakeType::PING;
                }
                break;
            }
        case HandshakeType::PING:
            {
                bytesRecv = recv(listen_fd_, buffer, 1024, 0);
                auto remessage_ptr=Paraser::parse_re_message(buffer,bytesRecv);
                if(remessage_ptr->type==ReMessageType::PONG){
                    vector<string> vct{"REPLCONF","listening-port",to_string(port)};
                    message=array_string(vct);
                    if (send(listen_fd_, message.c_str(),message.size(), 0) != -1) {
                        handshake_type=HandshakeType::REPLCONF_ONE;
                    }else{
                        std::cerr << "Error: Failed to send message" << std::endl;
                        handshake_type=HandshakeType::ERROR;
                    }
                }else{
                    handshake_type=HandshakeType::ERROR;
                    break;
                }

                break;
            }
        case HandshakeType::REPLCONF_ONE:
        {
            bytesRecv = recv(listen_fd_, buffer, 1024, 0);
            auto remessage_ptr=Paraser::parse_re_message(buffer,bytesRecv);
            if(remessage_ptr->type==ReMessageType::OK){
                vector<string> vct={"REPLCONF","capa","psync2"};
                message=array_string(vct);
                if (send(listen_fd_, message.c_str(),message.size(), 0) != -1) {
                    handshake_type=HandshakeType::REPLCONF_TWO;
                }else{
                    std::cerr << "Error: Failed to send message" << std::endl;
                    handshake_type=HandshakeType::ERROR;
                }
            }else{
                handshake_type=HandshakeType::ERROR;
            }
            break;
        }
        case HandshakeType::REPLCONF_TWO:
        {
            bytesRecv = recv(listen_fd_, buffer, 1024, 0);
            auto remessage_ptr=Paraser::parse_re_message(buffer,bytesRecv);
            if(remessage_ptr->type==ReMessageType::OK){
                vector<string> vct={"PSYNC","?","-1"};
                message=array_string(vct);
                if (send(listen_fd_, message.c_str(),message.size(), 0) != -1) {
                    handshake_type=HandshakeType::PSYNC;
                }else{
                    std::cerr << "Error: Failed to send message" << std::endl;
                    handshake_type=HandshakeType::ERROR;
                }
            }else{
                handshake_type=HandshakeType::ERROR;
            }
            break;
        }
        case HandshakeType::PSYNC:{
            bytesRecv = recv(listen_fd_, buffer, 1024, 0);
            auto remessage_ptr=Paraser::parse_re_message(buffer,bytesRecv);
            if(remessage_ptr->type==ReMessageType::UNKNOWN){
                handshake_type=HandshakeType::FINAL;
            }else{
                handshake_type=HandshakeType::ERROR;
            }
        }
        default:
            break;
        }
    }
    if(handshake_type==HandshakeType::FINAL){
        cout<<"Connected to master "<<master_ip<<":"<<master_port<<endl;
    }else{
        cout<<"Failed to connect to master "<<master_ip<<":"<<master_port<<endl;
    }
    
}

void RedisMaster::init() {
    this->redis_type=RedisType::REDIS_MASTER;
    init_socket();
};